package com.navinfo.platform.test;

import com.alibaba.fastjson.JSON;
import com.navinfo.opentsp.platform.location.protocol.gbv6.hangzhou.LCHangzhouDischarge;
import com.navinfo.platform.bean.IncidentAlermData;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

public class testKafkaClient_driving {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "10.30.50.121:9092");//you should write specific ip address rather than localhost
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");//StringSerializer/IntegerSerializer/or other self-defined Serializer.
//        properties.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("group.id", "MyGroup1");
        properties.put("auto.offset.reset", "earliest");
        Consumer<String,String> consumer  = new KafkaConsumer<String,String>(properties);
        consumer.subscribe(Arrays.asList("trip_statistic"));
        List<IncidentAlermData> incidentAlermDataList= new ArrayList<IncidentAlermData>();
        while(true){
            ConsumerRecords<String,String> records = consumer.poll(100);
            for(ConsumerRecord re:records){
                try {
                    IncidentAlermData incidentAlermData = JSON.parseObject((String)re.value(), IncidentAlermData.class);
//                    System.out.println(incidentAlermData.toString());
                    incidentAlermDataList.add(incidentAlermData);
                    if (incidentAlermData.getTid()==15588889999L){
                        System.out.println("incidentAlermData.toString()");
//                        incidentAlermDataList.add(incidentAlermData);
                    }

                }catch (Exception e){
                   e.printStackTrace();
                }
            }
            System.out.println(incidentAlermDataList.toString());
        }
    }
}